package com.im.business.api.group;

import com.alibaba.fastjson.JSONObject;
import com.im.business.api.group.entity.GroupUser;
import com.im.business.api.group.mapper.GroupUserMapper;
import com.im.common.ImConstant;
import com.im.common.SessionInfo;
import com.im.common.util.SnowFlakeUtil;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @Created leijun
 * @Date 2021/10/16 8:28 上午
 */
public class Application {

    private static SqlSessionFactory sqlSessionFactory;

    public static void main(String[] args) throws IOException, InterruptedException {
        SnowFlakeUtil snowFlakeUtil = new SnowFlakeUtil(ImConstant.DATACENTER_ID, ImConstant.MACHINE_ID);

        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaConsumer<String, String> consumer = kafkaManager.getConsumer();
        consumer.subscribe(Arrays.asList("group_message"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                String topic = record.topic();
                System.out.println("【业务层】接收到kafka topic【"+topic+"】消息："+record.value());
                JSONObject message = JSONObject.parseObject(record.value().toString());
                if (topic.equals("group_message")) {
                    Long messageId = snowFlakeUtil.nextId();
                    sendGroupMessageResponse(message, messageId);
                    pushGroupMessage(message, messageId);
                }
            }
            Thread.sleep(1000);
        }
    }

    /**
     * 响应群聊消息
     * @param message
     * @param messageId
     */
    private static void sendGroupMessageResponse(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("groupId", message.getLong("groupId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("group_message_response", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 推送群聊消息
     * @param message
     * @param messageId
     */
    private static void pushGroupMessage(JSONObject message, Long messageId) {
        SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            Long groupId = message.getLong("groupId");
            GroupUserMapper groupUserMapper = sqlSession.getMapper(GroupUserMapper.class);
            List<GroupUser> groupUsers = groupUserMapper.selectByGroup(groupId);
            KafkaProducer producer = KafkaManager.getProducer();
            for (GroupUser groupUser : groupUsers) {
                if (isUserOnline(groupUser.getUserId())) {
                    JSONObject messageResponse = new JSONObject();
                    messageResponse.put("senderId", message.getString("senderId"));
                    messageResponse.put("receiverId", groupUser.getUserId());
                    messageResponse.put("groupId", groupId);
                    messageResponse.put("messageId", messageId);
                    messageResponse.put("content", message.getString("content"));
                    messageResponse.put("timestamp", message.getLong("timestamp"));
                    ProducerRecord producerRecord = new ProducerRecord("push_group_message", messageResponse.toJSONString());
                    producer.send(producerRecord);
                } else {
                    pushOfflineMessage(message, messageId);
                }
            }
        }  finally {
            sqlSession.close();
        }
    }

    /**
     * 推送群聊的离线消息
     * @param message
     * @param messageId
     */
    private static void pushOfflineMessage(JSONObject message, Long messageId) {
        message.put("messageId", messageId);
        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaProducer kafkaProducer = kafkaManager.getProducer();
        kafkaProducer.send(new ProducerRecord("group_offline_message", message.toJSONString()));
    }

    /**
     * 用户是否在线
     * @param userId
     * @return
     */
    private static boolean isUserOnline(String userId) {
        String receiverSessionKey = "session_"+userId;
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(receiverSessionKey), SessionInfo.class);
        return sessionInfo == null ? false : true;
    }

}
